package cn.ivast.rxjava2test;

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.view.View;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import cn.ivast.rxjava2test.bean.LoginRequest;
import cn.ivast.rxjava2test.bean.LoginResponse;
import cn.ivast.rxjava2test.bean.RegisterRequest;
import cn.ivast.rxjava2test.bean.RegisterResponse;
import cn.ivast.rxjava2test.bean.UserBaseInfoRequest;
import cn.ivast.rxjava2test.bean.UserBaseInfoResponse;
import cn.ivast.rxjava2test.bean.UserExtraInfoRequest;
import cn.ivast.rxjava2test.bean.UserExtraInfoResponse;
import cn.ivast.rxjava2test.bean.UserInfo;
import cn.ivast.rxjava2test.utils.RetrofitUtil;
import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;

/**
 * RxJava2基本使用
 */
public class MainActivity extends AppCompatActivity {

    public static final String TAG = "MainActivity";
    private Disposable disposable;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        //基本使用
        findViewById(R.id.btn1).setOnClickListener(view -> fun1());

        //线程控制
        findViewById(R.id.btn2).setOnClickListener(view -> fun2());

        //网络请求
        findViewById(R.id.btn3).setOnClickListener(view -> fun3());

        //map操作符
        findViewById(R.id.btn4).setOnClickListener(view -> fun4());

        //flatMap
        findViewById(R.id.btn5).setOnClickListener(view -> fun5());

        //map操作符应用:依赖网络请求
        findViewById(R.id.btn6).setOnClickListener(view -> fun6());

        //zip操作符
        findViewById(R.id.btn7).setOnClickListener(view -> fun7());

        //zip操作符:模拟一个页面依赖多个接口数据
        findViewById(R.id.btn8).setOnClickListener(view -> fun8());

        //背压前提案例
        findViewById(R.id.btn9).setOnClickListener(view -> fun9());

        //背压前提案例2
        findViewById(R.id.btn10).setOnClickListener(view -> fun10());

        //Flowable基本使用
        findViewById(R.id.btn11).setOnClickListener(view -> fun11());

        findViewById(R.id.btn11).setOnLongClickListener(view -> {
            request(1);
            return true;
        });

        //interval
        findViewById(R.id.btn12).setOnClickListener(view -> fun12());

        //Flowable使用
        findViewById(R.id.btn13).setOnClickListener(view -> fun13());


    }

    private void fun13() {
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                Log.e(TAG, "before emit, requested = " + emitter.requested());
                Log.e(TAG, "emit 1");
                emitter.onNext(1);
                Log.e(TAG, "after emit 1, requested = " + emitter.requested());

                Log.e(TAG, "emit 2");
                emitter.onNext(2);
                Log.e(TAG, "after emit 2, requested = " + emitter.requested());

                Log.e(TAG, "emit 3");
                emitter.onNext(3);
                Log.e(TAG, "after emit 3, requested = " + emitter.requested());

                Log.e(TAG, "emit complete");
                emitter.onComplete();

            }
        },BackpressureStrategy.ERROR)
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        Log.e(TAG, "onSubscribe");
                        subscription = s;
                        s.request(10);
                        s.request(100);
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e(TAG, "onNext "+integer);
                    }

                    @Override
                    public void onError(Throwable t) {
                        Log.e(TAG, "onError");
                    }

                    @Override
                    public void onComplete() {
                        Log.e(TAG, "onComplete");
                    }
                });
    }

    /**
     * interval
     */
    private void fun12() {
        Flowable.interval(1,TimeUnit.MILLISECONDS)
                .onBackpressureDrop()
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Long>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        Log.e(TAG, "onSubscribe");
                        subscription = s;
                        s.request(Long.MAX_VALUE);
                    }

                    @Override
                    public void onNext(Long aLong) {
                        Log.e(TAG, "onNext: " + aLong);
                        try {
                            //延时1秒
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }

                    @Override
                    public void onError(Throwable t) {
                        Log.e(TAG, "onError: ");
                    }

                    @Override
                    public void onComplete() {
                        Log.e(TAG, "onComplete: ");
                    }
                });
    }

    Subscription subscription = null;

    public void request(long n) {
        //在外部调用request请求上游
        subscription.request(128);
    }

    /**
     * Flowable基本使用
     */
    private void fun11() {
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                for (int i = 0; i<10000; i++) {
//                    Log.e(TAG, "emit "+i);
                    emitter.onNext(i);
                }
//                Log.e(TAG, "emit onComplete");
//                emitter.onComplete();
            }
        }, BackpressureStrategy.LATEST)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        Log.e(TAG, "onSubscribe");
                        //注意这句代码
                        s.request(128);
                        subscription = s;
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e(TAG, "onNext: " + integer);
                    }

                    @Override
                    public void onError(Throwable t) {
                        Log.e(TAG, "onError: ", t);
                    }

                    @Override
                    public void onComplete() {
                        Log.e(TAG, "onComplete");
                    }
                });
    }

    /**
     * 背压前提案例2
     */
    private void fun10() {
        Observable.create((ObservableOnSubscribe<Integer>) emitter -> {
            for (int i = 0; ; i++) {
                emitter.onNext(i);
                Thread.sleep(2000);
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(integer -> Log.e(TAG, integer+""));
    }

    /**
     * 背压前提案例
     */
    private void fun9() {
        Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                for (int i = 0; ; i++) {
                    emitter.onNext(i);
                    Thread.sleep(2000);
                }
            }
        }).subscribeOn(Schedulers.io());

        Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                emitter.onNext("A");
            }
        }).subscribeOn(Schedulers.io());

        Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
            @Override
            public String apply(Integer integer, String s) throws Exception {
                return integer + s;
            }
        }).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.e(TAG, "accept");
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        Log.e(TAG, "Throwable");
                    }
                });
    }

    /**
     * zip操作符:模拟一个页面依赖多个接口数据
     */
    private void fun8() {
        Observable<UserBaseInfoResponse> observable1 = RetrofitUtil.create().getUserBaseInfo(new UserBaseInfoRequest()).subscribeOn(Schedulers.io());
        Observable<UserExtraInfoResponse> observable2 = RetrofitUtil.create().getUserExtraInfo(new UserExtraInfoRequest()).subscribeOn(Schedulers.io());
        Observable.zip(observable1, observable2, new BiFunction<UserBaseInfoResponse, UserExtraInfoResponse, UserInfo>() {
            @Override
            public UserInfo apply(UserBaseInfoResponse userBaseInfoResponse, UserExtraInfoResponse userExtraInfoResponse) throws Exception {
                return new UserInfo(userBaseInfoResponse, userExtraInfoResponse);
            }
        }).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<UserInfo>() {
                    @Override
                    public void accept(UserInfo userInfo) throws Exception {
                        //得到两个接口的数据
                    }
                });
    }

    /**
     * zip操作符简单使用
     */
    private void fun7() {
        Observable<Integer> observable1 = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.e(TAG, "emitter-->1");
                emitter.onNext(1);
                Thread.sleep(1000);

                Log.e(TAG, "emitter-->2");
                emitter.onNext(2);
                Thread.sleep(1000);

                Log.e(TAG, "emitter-->3");
                emitter.onNext(3);
                Thread.sleep(1000);

                Log.e(TAG, "emitter-->4");
                emitter.onNext(4);

                Log.e(TAG, "emitter--onComplete1");
                emitter.onComplete();
            }
        }).subscribeOn(Schedulers.io());

        Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                Log.e(TAG, "emitter-->A");
                emitter.onNext("A");
                Thread.sleep(1000);

                Log.e(TAG, "emitter-->B");
                emitter.onNext("B");
                Thread.sleep(1000);

                Log.e(TAG, "emitter-->C");
                emitter.onNext("C");
                Thread.sleep(1000);

                Log.e(TAG, "emitter--onComplete2");
                emitter.onComplete();
            }
        }).subscribeOn(Schedulers.io());

        Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {
            @Override
            public String apply(Integer integer, String s) throws Exception {
                Log.e(TAG, "----------" + Thread.currentThread().getName());
                return integer + s;
            }
        }).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.e(TAG, "onSubscribe");
                    }

                    @Override
                    public void onNext(String s) {
                        Log.e(TAG, "onNext-->" + s);
                        Log.e(TAG, "-----onNext-----" + Thread.currentThread().getName());
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.e(TAG, "onError");
                    }

                    @Override
                    public void onComplete() {
                        Log.e(TAG, "onComplete");
                    }
                });
    }

    /**
     * map操作符:依赖网络请求
     */
    private void fun6() {
        RetrofitUtil.create()
                .register(new RegisterRequest())
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext(new Consumer<RegisterResponse>() {
                    @Override
                    public void accept(RegisterResponse registerResponse) throws Exception {
                        //根据注册结果做某些操作
                    }
                }).observeOn(Schedulers.io())
                .flatMap(new Function<RegisterResponse, ObservableSource<LoginResponse>>() {
                    @Override
                    public ObservableSource<LoginResponse> apply(RegisterResponse registerResponse) throws Exception {
                        return RetrofitUtil.create()
                                .login(new LoginRequest());
                    }
                }).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<LoginResponse>() {
                    @Override
                    public void accept(LoginResponse loginResponse) throws Exception {
                        //登录成功
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        //登录出错
                    }
                });
    }

    /**
     * flatMap
     */
    private void fun5() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).concatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("I am value = " + integer);
                }
                return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.e(TAG, s);
            }
        });
    }

    /**
     * map操作符
     */
    private void fun4() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return "This is value = " + integer;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.e(TAG, s);
            }
        });
    }

    /**
     * 网络请求
     */
    private void fun3() {
        RetrofitUtil.create()
                .login(new LoginRequest())
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<LoginResponse>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                    }

                    @Override
                    public void onNext(LoginResponse loginResponse) {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    /**
     * 线程控制
     */
    private void fun2() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.e(TAG, "emitter-->" + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onComplete();
            }
        }).subscribeOn(Schedulers.newThread())
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "accept-->" + Thread.currentThread().getName());
                    }
                })
                .observeOn(Schedulers.io())
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "accept-->" + Thread.currentThread().getName());
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "onNext-->" + Thread.currentThread().getName());
                    }
                });
    }


    /**
     * 基本使用
     */
    private void fun1() {
        //创建一个上游
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.e(TAG, "emitter 1");
                emitter.onNext(1);
                Log.e(TAG, "emitter 2");
                emitter.onNext(2);
                Log.e(TAG, "emitter 3");
                emitter.onNext(3);
                Log.e(TAG, "emitter onComplete");
                emitter.onComplete();
                Log.e(TAG, "emitter 4");
                emitter.onNext(4);
                Log.e(TAG, "emitter 5");
                emitter.onNext(5);
            }
        });

        Observer<Integer> observer = new Observer<Integer>() {

            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "onSubscribe");
                disposable = d;
            }

            @Override
            public void onNext(Integer value) {
                Log.e(TAG, "onNext-->" + value);
                if (value == 2) {
                    Log.e(TAG, "onNext-->dispose()");
                    disposable.dispose();
                }
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError");
            }

            @Override
            public void onComplete() {
                Log.e(TAG, "onComplete");
            }
        };

        observable.subscribe(observer);

    }
}
